# producer.py
import json
import random
import time
from kafka import KafkaProducer

# Kafka 配置
bootstrap_servers = ['kafka1:9092', 'kafka2:9093', 'kafka3:9094']
topic = 'user-behavior'

# 创建生产者实例
producer = KafkaProducer(
    bootstrap_servers=bootstrap_servers,
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',                  # 等待所有 ISR 副本确认
    retries=3,                   # 重试次数
    max_in_flight_requests_per_connection=1,  # 确保消息按顺序发送且不丢失
    delivery_timeout_ms=30000,   # 发送超时时间
    enable_idempotence=True      # 启用幂等性（避免重复发送）
)

# 模拟用户行为数据
def generate_user_event():
    user_ids = [f"user_{i}" for i in range(1, 11)]
    actions = ["click", "view", "add_to_cart", "purchase", "logout"]
    
    return {
        "user_id": random.choice(user_ids),
        "action": random.choice(actions),
        "timestamp": int(time.time()),
        "product_id": random.randint(100, 200)
    }

if __name__ == "__main__":
    print("Kafka 生产者已启动，开始发送用户行为数据...")
    while True:
        event = generate_user_event()
        print(f"发送事件: {event}")  # 打印发送的 event
        future = producer.send(topic, event)
        try:
            record_metadata = future.get(timeout=10)  # 等待发送结果
            print(f"消息发送成功: {record_metadata.topic} - {record_metadata.partition}")
        except Exception as e:
            print(f"消息发送失败: {e}")
        time.sleep(5)  # 每5秒发送一条